'''

数据库执行公共函数

'''
import os
import time

from testcase.utils.ComThread import ComThread
from yat.test import Node
from yat.test import macro

from .Constant import Constant
from .Logger import Logger


class CommonSH:
    def __init__(self, node_name='dbuser'):
        # 通过root用户还是数据库安装的用户执行脚本。 数据库部署用户默认使用dbuser（见 conf/nodes.yml）
        self.DB_INSTANCE_PATH = macro.DB_INSTANCE_PATH
        self.DB_ENV_PATH = macro.DB_ENV_PATH
        self.log = Logger()
        self.Constant = Constant()
        self.node = Node(node=node_name)

    def startDbCluster(self, get_detail=False, env=None):
        """
        gs_om工具启动集群
        :param env:指定环境变量路径
        :param get_detail:显示详细回显信息
        :return:启动结果
        """
        env = self.DB_ENV_PATH if not env else env
        start_cmd = f'source {env};gs_om -t start'
        self.log.info(start_cmd)
        msg = self.node.sh(start_cmd).result()
        self.log.info(msg)
        time.sleep(10)
        if get_detail:
            return msg
        else:
            return msg.find(self.Constant.START_SUCCESS_MSG) > -1

    def stopDbCluster(self, max_time='1800', command='', get_detail=False,
                      env=None):
        """
        gs_om工具停止集群
        :param env:指定环境变量路径
        :param max_time:超时时间
        :param command:其他参数，如-h HOSTNAME
        :param get_detail:显示详细回显信息
        :return:停止结果
        """
        env = self.DB_ENV_PATH if not env else env
        cmd = f'source {env};gs_om --time-out={max_time} -t stop {command}'
        self.log.info(cmd)
        stopmsg = self.node.sh(cmd).result()
        self.log.info(stopmsg)
        time.sleep(3)
        if get_detail:
            return stopmsg
        else:
            return stopmsg.find(self.Constant.STOP_SUCCESS_MSG) > -1

    def om_restart(self, env=None, max_wait_time='1800', other_param='',
                   get_detail=False):
        """
        gs_om工具重启数据库
        :param env: 指定环境变量路径
        :param max_wait_time: 超时时间
        :param other_param: 其他参数，如-h HOSTNAME
        :param get_detail: 显示详细回显信息
        :return:
        """
        env = self.DB_ENV_PATH if not env else env
        cmd = f'source {env};gs_om -t restart --time-out={max_wait_time} ' \
              f'{other_param}'
        self.log.info(cmd)
        res = self.node.sh(cmd).result()
        self.log.info(res)
        time.sleep(10)
        if get_detail:
            return res
        else:
            return res.find(self.Constant.STOP_SUCCESS_MSG) > -1 and \
                   res.find(self.Constant.START_SUCCESS_MSG) > -1

    def getDbClusterStatus(self, type='info', env_path=macro.DB_ENV_PATH):
        '''
        type="info":返回数据库集群状态查询的日志信息
        type="status":返回数据库集群状态
        '''
        if type == "status":
            body = '''source {};gs_om -t status --detail'''.format(env_path)
            res = self.node.sh(body).result()
            self.log.info(res)
            if "stopped" in res or "repair" in res or "Unknown" in res \
                    or "Coredump" in res:
                return False
            else:
                return True
        elif type == 'detail':
            cmd = 'source {};gs_om -t status --detail'.format(env_path)
            self.log.info(cmd)
            return self.node.sh(cmd).result()
        else:
            cmd = 'source {};gs_om -t status'.format(env_path)
            self.log.info(cmd)
            return self.node.sh(cmd).result()

    def startDbInstance(self, mode="primary"):
        '''
        启动数据库实例
        :param mode: primary or standby
        '''
        self.log.info("==startDbInstance==")
        msg = self.node.sh(
            'source {};gs_ctl start -D {} -M {}'.format(self.DB_ENV_PATH, self.DB_INSTANCE_PATH, mode)).result()
        self.log.info(msg)
        time.sleep(10)
        return msg

    def stopDbInstance(self):
        '''
        停止数据库实例
        '''
        self.log.info("==stopDbInstance==")
        msg = self.node.sh('source {};gs_ctl stop -D {}'.format(self.DB_ENV_PATH, self.DB_INSTANCE_PATH)).result()
        self.log.info(msg)
        time.sleep(3)
        return msg

    def getDbInstanceStatus(self):
        '''
        查询数据库状态
        返回： 正常运行 True  其他： False
        '''
        self.log.info("==get Db Instance Status==")
        msg = self.node.sh('source {};gs_ctl status -D {}'.format(self.DB_ENV_PATH, self.DB_INSTANCE_PATH)).result()
        self.log.info(msg)
        return msg.find(self.Constant.INSTANCE_RUNNING) > -1

    def reloadDbConfig(self):
        '''
        重新加载数据库配置文件 pg_hba.conf  postgresql.conf
        '''
        self.log.info("==reload Db Instance Conf==")
        msg = self.node.sh('source {};gs_ctl reload -D {}'.format(self.DB_ENV_PATH, self.DB_INSTANCE_PATH)).result()
        self.log.info(msg)
        time.sleep(5)
        return msg

    def executDbSql(self, sql, dbname=None, cmd='',
                    env_path=macro.DB_ENV_PATH):
        """
        使用gsql的方式调用数据库sql语句
        :param sql: sql语句
        :param dbname: 执行的目标数据库
        :param env_path: 环境变量
        :param cmd: 其他参数，例如-U -W -h等参数拼接
        :return: sql语句执行结果
        """
        if dbname is None:
            database_name = self.node.db_name
        else:
            database_name = dbname
        sql_cmd = f'source {env_path}; ' \
            f'gsql -d {database_name} ' \
            f'-p {self.node.db_port} ' \
            f'-r ' \
            f'{cmd} ' \
            f'-c "{sql}" '
        self.log.info(sql_cmd)
        sql_result = self.node.sh(sql_cmd).result()
        self.log.info(sql_result)
        return sql_result

    def executeGsctl(self, command, assert_flag, param='', get_detail=False):
        '''
        gs_ctl工具
        示例：
        command='build', assert_flag=self.Constant.REBUILD_SUCCESS_MSG, param='-b full'
        调用：
        self.sh.executeGsctl('build', self.Constant.REBUILD_SUCCESS_MSG, '-b full')
        得到：
        gs_ctl build -D /opt/gaussdb/dn1 -b full
        '''
        self.log.info("==gs_ctl start execute==")
        time_out = ""
        if 'build' == command:
            time_out = '-t 600'
        rebuild_cmd = f'''
               source {self.DB_ENV_PATH}; 
               gs_ctl {command} -D {self.DB_INSTANCE_PATH} {param} {time_out};
               '''
        self.log.info(rebuild_cmd)
        show_msg = self.node.sh(rebuild_cmd).result()
        self.log.info(show_msg)
        time.sleep(5)

        if get_detail:
            return show_msg
        else:
            return show_msg.find(assert_flag) > -1

    def executeGsguc(self, command, assert_flag, param, node_name='all',
                     get_detail=False, single=False, dn_path='',
                     pghba_param='', env_path=macro.DB_ENV_PATH):
        """
        gs_guc工具
        :param command: 例如'check'
        :param assert_flag: 例如self.Constant.GSGUC_SUCCESS_MSG
        :param param: 例如'max_connections'
        :param node_name: 节点名
        :param get_detail: True返回详细信息，False返回bool值
        :param single: True代表仅在本节点执行gs_guc命令，False在所有节点执行gs_guc命令
        :param dn_path: 可指定数据目录
        :param pghba_param: 适配修改pg_hba
        :return: 返回执行结果
        """
        if not dn_path:
            dn_path = self.DB_INSTANCE_PATH
        self.log.info("----gs_guc start execute----")
        if single:
            if pghba_param:
                gs_guc_cmd = f'source {env_path}; ' \
                    f'gs_guc {command} ' \
                    f'-D {dn_path} ' \
                    f'-h "{pghba_param}";'
            else:
                gs_guc_cmd = f'source {env_path}; ' \
                    f'gs_guc {command} ' \
                    f'-D {dn_path} ' \
                    f'-c "{param}";'
        else:
            if pghba_param:
                gs_guc_cmd = f'source {env_path}; ' \
                    f'gs_guc {command} ' \
                    f'-N {node_name} ' \
                    f'-D {dn_path} ' \
                    f'-h "{pghba_param}";'
            else:
                gs_guc_cmd = f'source {env_path}; ' \
                    f'gs_guc {command} ' \
                    f'-N {node_name} ' \
                    f'-D {dn_path} ' \
                    f'-c "{param}";'
        self.log.info(gs_guc_cmd)
        show_msg = self.node.sh(gs_guc_cmd).result()
        self.log.info(show_msg)
        if command == 'reload':
            time.sleep(5)
        if get_detail:
            return show_msg
        else:
            return show_msg.find(assert_flag) > -1

    def show_param(self, guc_param):
        """
        function:查看guc参数
        :param guc_param: guc参数
        :return: 返回参数结果
        """
        check_default = f'show {guc_param};'
        check_result = self.executDbSql(check_default)
        check_msg = check_result.splitlines()[-2].strip()
        self.log.info(check_msg)
        return check_msg

    def exec_refresh_conf(self, detail=False):
        '''
        执行refreshconf
        :return:执行成功返回True，否则返回False
        '''
        self.log.info("==refreshconf start execute==")
        refresh_cmd = f'''source {self.DB_ENV_PATH}; gs_om -t refreshconf '''
        self.log.info(refresh_cmd)
        refresh_msg = self.node.sh(refresh_cmd).result()
        self.log.info(refresh_msg)
        if detail:
            return refresh_msg
        else:
            status = refresh_msg.find(self.Constant.REFRESH_SUCCESS_MSG) > -1
            time.sleep(5)
            return status

    def cycle_exec_sql(self, sql, count, log=False, dbname=None,
                       get_detail=False):
        """
        循环执行sql语句count次
        :param sql: 要执行的sql语句
        :param count: 执行次数
        :param log: 是否打印每次执行日志
        :param dbname: 执行sql的库
        :param get_detail: 是否获取循环执行的所有结果
        :return: 执行完成返回True
        """
        db_name = dbname if dbname else self.node.db_name
        res_list = []
        run_flag = True
        try:
            shell = f'source {self.DB_ENV_PATH}; ' \
                f'gsql -d {db_name} ' \
                f'-p {self.node.db_port} ' \
                f'-r ' \
                f'-c "{sql}" '
            self.log.info(shell)
            for i in range(count):
                if log:
                    self.log.info("循环执行第" + str(i) + "次")
                msg = self.node.sh(shell).result()
                if log:
                    self.log.info(msg)
                if get_detail:
                    res_list.append(msg)
        except Exception as e:
            self.log.error('执行失败！')
            self.log.info(e)
            run_flag = False
        else:
            self.log.info('执行成功！')
        if get_detail:
            return res_list
        else:
            return run_flag

    def restoreFile(self, filename, cmd='-c', get_detail=True, dbname=None):
        '''
        :param file_name: 将输出发送至指定文件或目录
        :param cmd: 自定义增加一些参数，例如:
        1、默认-c 在重新创建数据库对象前，清理（删除）已存在于将要还原的数据库中的数据库对象。
        2、其他参数设定等（连接参数或者转储参数等，可参考工具参考文档）
        :param get_detail，是否return详细返回信息
        :return: 根据get_detail参数返回相应结果
        '''
        if dbname is None:
            dbname = self.node.db_name
        restore_cmd = '''
               source {source_path};
               gs_restore -U '{username}' -W '{password}' -p {port} -d {dbname} '{file_name}' {cmd}'''.format(
            source_path=self.DB_ENV_PATH,
            dbname=dbname,
            port=self.node.db_port,
            username=self.node.db_user,
            password=self.node.db_password,
            file_name=filename,
            cmd=cmd)
        self.log.info(restore_cmd)
        restore_msg = self.node.sh(restore_cmd).result()
        self.log.info(restore_msg)
        if get_detail:
            return restore_msg
        else:
            return restore_msg.find(self.Constant.RESTORE_SUCCESS_MSG) > -1

    def dumpFile(self, filename, cmd='-F t', dbname=None, get_detail=True):
        '''
        :param filename: 将输出发送至指定文件或目录
        :param cmd: 自定义增加一些参数，例如:
        1、默认指定输出格式：-F t (输出格式类型：1、p|plain：输出一个文本SQL脚本文件；2、c|custom：输出一个自定义格式的归档；3、d|directory：该格式会创建一个目录；4、t|tar：输出一个tar格式的归档形式)
        2、其他参数设定等（连接参数或者转储参数等，可参考工具参考文档）
        :param dbname: 可传入数据库名
        :param get_detail，是否return详细返回信息
        :return: 根据get_detail参数返回相应结果
        '''
        if dbname is None:
            dbname = self.node.db_name

        dump_cmd = '''
               source {source_path};
               gs_dump {dbname} -p {port} -f {file_name} {cmd}'''.format(
            source_path=self.DB_ENV_PATH,
            dbname=dbname,
            port=self.node.db_port,
            file_name=filename,
            cmd=cmd)
        self.log.info(dump_cmd)
        dump_msg = self.node.sh(dump_cmd).result()
        self.log.info(dump_msg)
        flag = 'dump database ' + dbname + ' successfully'
        if get_detail:
            return dump_msg
        else:
            return dump_msg.find(flag) > -1

    def buildStandby(self, mode='-b full'):
        dump_cmd = f"source {self.DB_ENV_PATH};" \
            f"gs_ctl build -D {self.DB_INSTANCE_PATH} {mode} -t 600"
        self.log.info(dump_cmd)
        build_msg = self.node.sh(dump_cmd).result()
        self.log.info(build_msg)

        time.sleep(5)

        return build_msg

    def nosocketConnect(self):
        no_socket_connect_cmd = f'source {self.DB_ENV_PATH};gsql -h {self.node.db_host} -d {self.node.db_name} -p {self.node.db_port} -U "{self.node.db_user}" -W "{self.node.db_password}" -c \'\q\''
        self.log.info(no_socket_connect_cmd)
        no_socket_connect_msg = self.node.sh(no_socket_connect_cmd).result()
        self.log.info(no_socket_connect_msg)
        return no_socket_connect_msg

    def socket_connect(self, file_path):
        socket_connect_cmd = f'source {self.DB_ENV_PATH};gsql -d {self.node.db_name} -p {self.node.db_port} -h {file_path} -c \'\q\''
        self.log.info(socket_connect_cmd)
        socket_connect_msg = self.node.sh(socket_connect_cmd).result()
        self.log.info(socket_connect_msg)
        return socket_connect_msg

    def get_sync_status(self):
        sender_replay_location = ''
        receiver_replay_location = ''
        query_msg = self.executeGsctl('query', '', get_detail=True)
        test_list = query_msg.strip().splitlines()
        for ts in test_list:
            if 'sender_replay_location' in ts:
                sender_replay_location = ts.split(':')[-1].strip()
            elif 'receiver_replay_location' in ts:
                receiver_replay_location = ts.split(':')[-1].strip()

        return sender_replay_location, receiver_replay_location

    def check_data_consistency(self):
        start_time = time.time()
        while True:
            sender_replay_location, receiver_replay_location = self.get_sync_status()
            if len(sender_replay_location) > 0 and len(receiver_replay_location) > 0:
                if sender_replay_location == receiver_replay_location:
                    self.log.info('sender_replay_location:' + sender_replay_location)
                    self.log.info('receiver_replay_location:' + receiver_replay_location)
                    return True
            time.sleep(10)
            end_time = time.time()
            # 设置超时退出，避免死循环
            if end_time - start_time > 1800:
                self.log.error('gs_ctl query 返回信息异常或数据同步仍未完成，请检查!')
                return False

    def build_fail_scp_crtfile(self):
        rebuild_cmd = f'''
                       source {self.DB_ENV_PATH}; 
                       gs_ctl build -D {self.DB_INSTANCE_PATH} -t 600;
                       '''
        self.log.info(rebuild_cmd)
        show_msg = self.node.sh(rebuild_cmd).result()
        self.log.info(show_msg)
        if show_msg.find(self.Constant.BUILD_FAIL_MSG2) > -1 or show_msg.find(
                self.Constant.BUILD_FAIL_MSG3) > -1 or show_msg.find(self.Constant.BUILD_FAIL_MSG4) > -1:
            self.log.info('*****start cp server.crt files*****')
            time.sleep(20)
            find_cmd = f"find {os.path.dirname(macro.DB_INSTANCE_PATH)} -type d -name 'sslcert'"
            self.log.info(find_cmd)
            om_path = self.node.sh(find_cmd).result()
            self.log.info(om_path)

            str_cp = r"\cp"
            edit_cmd = f'''cd {om_path.strip()}/om/;
                    {str_cp} server.key.rand server.key.cipher server.key server.crt cacert.pem {macro.DB_INSTANCE_PATH}'''
            self.node.sh(edit_cmd)
            self.log.info(edit_cmd)
            msg = self.executeGsctl('build', self.Constant.REBUILD_SUCCESS_MSG)
            self.log.info(msg)
            return msg
        elif not show_msg.find(self.Constant.REBUILD_SUCCESS_MSG) > -1:
            build_msg = self.executeGsctl('build', self.Constant.REBUILD_SUCCESS_MSG)
            self.log.info(build_msg)
            return build_msg
        return True

    def check_whether_need_switch(self, hostname):
        """
        判断主机是否需要switchover
        :param hostname:主机hostname
        :return:需要switchover返回True，不需要返回False
        """
        db_status = self.getDbClusterStatus('detail')
        self.log.info(db_status)
        if db_status.count('|') > 0:
            dest_str = db_status.splitlines()[-1].strip()
            dest_list = dest_str.split('|')
        else:
            dest_str = db_status.split('[ Datanode State ]')[-1].strip()
            dest_list = dest_str.splitlines()[2::]

        flag = False
        for status in dest_list:
            if hostname in status and 'Primary Normal' not in status:
                self.log.info(status)
                flag = True
        return flag

    def get_node_num(self):
        """
        判断备机数量
        :return:返回备机数量
        """
        db_status = self.getDbClusterStatus('detail')
        if db_status.count('|') > 0:
            dest_str = db_status.splitlines()[-1].strip()
            dest_list = dest_str.split('|')
        else:
            dest_str = db_status.splitlines()
            split_idx = 0
            for item_index, item in enumerate(dest_str):
                if '-----' in item:
                    split_idx = item_index
            dest_list = dest_str[split_idx + 1:]

        node_num = len(dest_list)
        self.log.info(dest_list)
        self.log.info('节点数量为：'+str(node_num))
        return node_num

    def check_whether_need_build(self):
        """
        判断备机是否需要重建
        :return:需要重建返回True，不需要重建返回False
        """
        node_num = self.get_node_num()
        db_status = self.getDbClusterStatus('detail')
        standby_normal_num = db_status.count('Standby Normal')
        self.log.info(f'node_num:{node_num}, '
                      f'standby_normal_num:{standby_normal_num}')
        self.log.info(db_status)
        if int(standby_normal_num) == int(node_num) - 1:
            return False
        else:
            return True

    def check_whether_need_wait(self):
        db_status = self.getDbClusterStatus('detail')
        self.log.info(db_status)
        if 'Catchup' in db_status or ('Standby Need repair' in db_status
                                      and '(WAL)' not in db_status):
            return True
        else:
            return False

    def check_whether_remain_dirty_page(self):
        start_time = time.time()
        sql_cmd = "select remain_dirty_page_num " \
                  "from  DBE_PERF.GLOBAL_PAGEWRITER_STATUS;"
        while True:
            time.sleep(30)
            msg = self.executDbSql(sql_cmd)
            self.log.info(msg)
            num = msg.splitlines()[-2].strip()
            if num == '0':
                return True

            end_time = time.time()
            if end_time - start_time > 1800:
                return False

    def gsql_restore(self, dump_file):
        '''
        function:使用gsql的方式调用数据库sql语句
        args: dump_file:需要导入的sql文件名称（需包含路径)
        return: gsql 导入文件所有打印信息

        '''
        shell = '''
        source {env};
        gsql -d {db_name} -p {port} -U '{user}' -W '{password}' -f '{dump_file}'
        '''.format(env=self.DB_ENV_PATH, db_name=self.node.db_name, port=self.node.db_port, user=self.node.db_user,
                   password=self.node.db_password, dump_file=dump_file)
        self.log.info(shell)
        msg = self.node.sh(shell).result()
        return msg

    def exec_pg_controldata(self, para_name_list):
        param_value_list = []
        shell_cmd = f"source {self.DB_ENV_PATH}; pg_controldata {self.DB_INSTANCE_PATH}"
        self.log.info(shell_cmd)
        result_lines = self.node.sh(shell_cmd).result()
        for result_line in result_lines.splitlines():
            for para_name in para_name_list:
                if para_name in result_line:
                    param_value_list.append(result_line.split()[-1].strip())

        return param_value_list

    def execu_gs_basebackup(self, backup_path, node_ip, node_port, cmd=''):
        """
        Function:执行gs_basebackup备份指令
        :param backup_path:备份文件存放路径
        :param node_ip:需要备份的节点ip信息
        :param node_port:需要备份节点数据库的port口信息
        :param cmd:其他参数，例如-X
        :return:指令执行打印信息
        """
        shell_cmd = f"source {macro.DB_ENV_PATH};gs_basebackup " \
            f"-D {backup_path} -h {node_ip} -p {node_port} {cmd} -v -t 3600"
        self.log.info(shell_cmd)
        result = self.node.sh(shell_cmd).result()
        self.log.info(result)
        return result

    def get_dirty_page_num(self):
        """
        Function:获取当前环境中脏页个数
        :return:脏页个数
        """
        sql_cmd = "select remain_dirty_page_num " \
                  "from DBE_PERF.GLOBAL_PAGEWRITER_STATUS;"
        msg = self.executDbSql(sql_cmd)
        self.log.info(msg)
        num = msg.splitlines()[-2].strip()
        return num

    def execu_gs_backup(self, param, host, backup_path, cmd):
        """
        Function:执行gs_backup并返回执行结果
        :param param:backup or restore
        :param host:-h参数
        :param backup_path:--backup-dir参数
        :param cmd:其他参数类似-l,--parameter等
        :return:执行后打印信息
        """
        cmd = f"source {self.DB_ENV_PATH};" \
            f"gs_backup -t {param} --backup-dir={backup_path} -h {host} {cmd}"
        self.log.info(cmd)
        result = self.node.sh(cmd).result()
        self.log.info(result)
        return result

    def check_location_consistency(self, node_type, node_num,
                                   max_wait_time=1800):
        """
        Function:判断主备数据是否同步
        :param ：node_type，传入主节点还是备节点，'primary'，'standby'
        :param ：node_num，集群个数，一主两备传入3，一主一备传入2
        :param ：max_wait_time，最多等待时长，单位：秒
        :return:返回True表明已经同步；False表明在规定时间内未同步
        """
        start_time = time.time()
        while True:
            query_msg = self.executeGsctl('query', '', get_detail=True)
            result_list = list()
            assert_flag = True
            status_flag = True
            for arg in query_msg.splitlines():
                if 'sender_replay_location' in arg or \
                        'receiver_replay_location' in arg:
                    result_list.append(arg.strip())
            result_set = set(result_list)
            if len(result_set) > 0:
                if node_type == 'primary':
                    if len(result_set) * (node_num - 1) != len(result_list):
                        self.log.info('主备未同步')
                        db_status = self.getDbClusterStatus('detail')
                        self.log.info(db_status)
                        if 'Standby Need repair(WAL)' in db_status or \
                                'stopped' in db_status:
                            self.log.error('存在无法同步情况，请检查!')
                            status_flag = False
                        time.sleep(10)
                        assert_flag = False
                value_list = list()
                if assert_flag:
                    for arg in result_list:
                        value_list.append(arg.split(':')[-1].strip())
                    if value_list.count(value_list[0]) == len(result_list):
                        self.log.info('主备已同步')
                        return True
                    else:
                        self.log.info('主备未同步')
                        time.sleep(10)
            if len(result_set) == 0:
                db_status = self.getDbClusterStatus('detail')
                self.log.info(db_status)
                if 'Standby Need repair(WAL)' in db_status or \
                        'stopped' in db_status:
                    self.log.error('存在无法同步情况，请检查!')
                    status_flag = False

            if not status_flag:
                return False

            end_time = time.time()
            # 设置超时退出，避免死循环
            if end_time - start_time > max_wait_time:
                self.log.error('返回信息异常或数据同步仍未完成，请检查!')
                return False

    def wait_cluster_connected(self):
        """
        function:等待数据库主节点恢复，变为可连接状态
        :return:1小时恢复为可连接True，否则False
        """
        cmd = f"source {macro.DB_ENV_PATH};gs_om -t status --all"
        for i in range(60):
            self.log.info(cmd)
            result = self.node.sh(cmd).result()
            self.log.info(result)
            lines = result.strip().splitlines()
            for line in lines:
                if 'instance_state' in line:
                    if 'Normal' in line:
                        self.log.info(line)
                        return True
                    else:
                        self.log.info(line)
                        break
            time.sleep(60)
        return False

    def get_standby_and_build(self):
        """
        function: 通过postgres.conf文件获取备机ip，ssh到备机后执行备机重建
        :return: 备机重建回显信息
        """
        conf_path = os.path.join(macro.DB_INSTANCE_PATH,
                                 macro.DB_PG_CONFIG_NAME)
        self.log.info('----获取备节点ip----')
        shell_cmd = f"cat {conf_path} | " \
            f"grep 'replconninfo' | " \
            f"grep -Ev '^#' | " \
            f"tr -s ' '| " \
            f"cut -d ' ' -f 7 | " \
            f"cut -d '=' -f 2"
        self.log.info(shell_cmd)
        msg = self.node.sh(shell_cmd).result()
        self.log.info(msg)
        standby_ip_list = msg.splitlines()

        self.log.info('----备机重建----')
        build_msg_list = list()
        for ip in standby_ip_list:
            shell_cmd = f'''ssh {ip} <<-EOF
                        source {macro.DB_ENV_PATH}
                        gs_ctl build -D {macro.DB_INSTANCE_PATH} -b full
                        exit\n''' + "EOF"
            self.log.info(shell_cmd)
            build_msg = self.node.sh(shell_cmd).result()
            build_msg_list.append(build_msg)

        return build_msg_list

    def check_cascade_standby_consistency(self):
        '''
        Function: 查询级联备同步情况
        :return: 同步返回True否则返回False
        '''
        cmd = f"source {macro.DB_ENV_PATH};gs_om -t status --all"
        self.log.info(cmd)
        result = self.node.sh(cmd).result()
        self.log.info(result)
        list_tmp = result.split('----------------------------------------')
        flg_cascade = 0
        flg_standby = 0
        sender_replay_location = ''
        receiver_replay_location = ''
        for i in range(len(list_tmp)-1):
            if 'Cascade Standby' in list_tmp[i]:
                detail_list = list_tmp[i].strip().splitlines()
                for j in range(len(detail_list)-1):
                    if 'receiver_replay_location' in detail_list[j]:
                        receiver_replay_location = \
                            detail_list[j].strip().split(':')[1]
                        self.log.info(f"receiver_replay_location "
                                      f"is {receiver_replay_location}")
                        flg_cascade = flg_cascade + 1
            elif ': Standby' in list_tmp[i]:
                detail_list = list_tmp[i].strip().splitlines()
                for j in range(len(detail_list) - 1):
                    if 'sender_replay_location' in detail_list[j]:
                        sender_replay_location = \
                            detail_list[j].strip().split(':')[1]
                        self.log.info(f"sender_replay_location "
                                      f"is {sender_replay_location}")
                        flg_standby = flg_standby + 1
        if flg_standby > 0 and flg_cascade > 0:
            if sender_replay_location == receiver_replay_location:
                return True
        else:
            return False

    def exec_expension(self, user, group, host, xml, para="-L", detail=False):
        '''
        执行gs_expension
        host :对应-h 表示待扩容节点
        para :默认为-L 可传入为其他值
        user :对应-U 数据库安装用户
        group :对应-G 数据库属主
        xml :对应-X 数据库xml文件
        detail:False则返回扩容成功或失败，True则返回扩容执行结果
        return:执行成功返回True，否则返回False 如果detail为真返回执行详细结果
        '''

        # 如果在备机上对集群进行安装 则需要再主机上解压安装包以执行扩容
        cmd = f"ls {macro.DB_SCRIPT_PATH}"
        self.log.info(cmd)
        result = self.node.sh(cmd).result()
        self.log.info(result)
        if "gs_expansion" not in result:
            cmd = f"cd {macro.DB_SCRIPT_PATH}/../;" \
                f"ls -al | grep openGauss-Package-bak"
            self.log.info(cmd)
            result = self.node.sh(cmd).result()
            self.log.info(result)
            if 'openGauss-Package-bak' in result:
                cmd = f"cd {macro.DB_SCRIPT_PATH}/../;" \
                    f"tar -zxvf {result.split()[-1].strip()} > /dev/null"
                self.log.info(cmd)
                result = self.node.sh(cmd).result()
                self.log.info(result)
            else:
                raise Exception("can't find package")
            result = self.node.sh(
                f"ls {macro.DB_SCRIPT_PATH}").result()
            if "gs_expansion" not in result:
                self.log.info("cat not find gs_expansion, Please check!")
                return False

        execute_cmd = f'''source {macro.DB_ENV_PATH};
                cd {macro.DB_SCRIPT_PATH};
                ./gs_expansion -U {user} \
                -G {group} \
                -h {host} -X {xml} {para}'''
        self.log.info(execute_cmd)
        result = self.node.sh(execute_cmd).result()
        self.log.info(result)
        time.sleep(5)
        # detail默认为False 返回扩容成功或失败 detail为True时返回扩容执行结果
        if detail:
            return result
        else:
            return result.find("Expansion Finish") > -1

    def exec_dropnode(self, drop_node, timeout=120):
        '''
        function: 执行减容操作
        drop_node: 被减容节点
        timeout: 超时时间
        :return: 减容打印信息
        '''
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                                    expect <<EOF
                                    set timeout {timeout}
                                    spawn gs_dropnode -U \
                                    {self.node.ssh_user} \
                                    -G {self.node.ssh_user} \
                                    -h {drop_node.ssh_host}
                                    expect "*drop the target node (yes/no)?*"
                                    send "yes\\n"
                                    expect eof\n''' + '''EOF'''
        self.log.info(execute_cmd)
        result = self.node.sh(execute_cmd).result()
        return result

    def exec_gs_sshexkey(self, script_path, *args, **kwargs):
        """
        gs_sshexkey工具使用
        :param script_path: 工具脚本存放路径
        :param args: host元组，如(10.10.10.10, 11.11.11.11)
        :param kwargs: 参数键值对，如 {'-f': 'test', '-l': '/home/test.txt'}
        :return: 回显信息
        """
        if not kwargs:
            raise Exception("参数不能为空!")
        host_str = ''
        del_str = ''
        if args:
            if kwargs.get('-f', ''):
                f_name = kwargs.get('-f')
            else:
                f_name = 'sshexkey_hosts'
            host_str = f'> {f_name}\n'
            for i in args:
                host_str += f'echo "{i}" >> {f_name}\n'
            del_str = f'rm -rf {f_name}'
        param_str = ''
        for k, v in kwargs.items():
            param_str += f'{k} {v}'
        cmd = f'cd {script_path}\n' if script_path else ''
        cmd += f'''{host_str}
            expect <<EOF
            set timeout 300
            spawn ./gs_sshexkey {param_str}
            expect {{{{
                "*(yes/no)?" {{{{ send "yes\r";exp_continue }}}}
                "*assword:" {{{{ send "{self.node.ssh_password}\r" }}}}
                "*]#" {{{{ send "\r" }}}}
            }}}}
            expect eof''' + f'\nEOF\n{del_str}'
        self.log.info(cmd)
        res = self.node.sh(cmd).result()
        self.log.info(res)
        return res

    def execute_generate(self, factor, prefix='subscription',
                         path='$GAUSSHOME/bin',
                         env_path=macro.DB_ENV_PATH):
        """
        执行gs_guc generate
        :param factor: 加密因子
        :param prefix: 前缀
        :param path: 生成路径
        :param env_path: env路径
        :return: 执行结果
        """
        cmd = f"source {env_path};" \
            f"gs_guc generate -S {factor} -D {path} -o {prefix}"
        self.log.info(cmd)
        result = self.node.sh(cmd).result()
        self.log.info(result)
        return result


class RestartDbCluster(object):
    def __init__(self, *args):
        self.log = Logger()
        self.Constant = Constant()
        self.shs = [CommonSH(n) for n in args]
        self.nodes = [Node(node=n) for n in args]
        self.root_nodes = [Node(node=n.replace('DbUser', 'Root')) for n in args]
        self.thread_list = list()

    def __call__(self, func):
        def wrapper(*args, **kwargs):
            self.log.info("-----验证数据库用户互信是否正常------")
            cmd = f'''source {macro.DB_ENV_PATH};
            gs_ssh -c "gs_ssh -c \"date\""'''
            gs_ssh_msg = self.nodes[0].sh(cmd).result()
            self.log.info(gs_ssh_msg)

            self.log.info("-----清理缓存------")
            clean_cmd = f"sync;" \
                f"echo 3 > /proc/sys/vm/drop_caches; " \
                f"ipcs -m|awk '$3~/^\[0-9\]+/{{{{print $0}}}}'|" \
                f"awk '{{{{print $2}}}}'|xargs -I {{{{}}}} ipcrm -m {{{{}}}}"
            self.log.info(clean_cmd)
            self.thread_list.clear()
            for rn in self.root_nodes:
                self.thread_list.append(ComThread(rn.sh, args=(clean_cmd,)))
            for t in self.thread_list:
                t.setDaemon(True)
                t.start()
            for lt in self.thread_list:
                lt.join(20)
                rs = lt.get_result()
                if rs:
                    self.log.info(rs.result())
                else:
                    self.log.error("清理缓存超时！")

            self.log.info("----查看是否有超时进程，如有则杀进程----")
            now_pid = os.getpid()
            self.log.info(now_pid)
            cmd = f"ps -ef|yat.runner --test-suite|grep -v grep|" \
                  f"grep -vw {now_pid}|awk '{{{{print $2}}}}'|" \
                  f"xargs -I {{}} kill -9 {{}}"
            result = os.popen(cmd)
            self.log.info(result.readlines())

            self.log.info("----重启集群----")
            stopmsg = self.shs[0].stopDbCluster()
            self.log.info(stopmsg)
            if not stopmsg:
                self.log.error('数据库停止失败，请检查！')

            start_msg = self.shs[0].startDbCluster()
            self.log.info(start_msg)
            if not start_msg:
                self.log.error('数据库启动失败，请检查！')

            self.log.info("----查看集群状态----")
            db_status = self.shs[0].getDbClusterStatus('detail')
            if db_status.count('Primary Normal') > 1:
                self.log.info('----多主时降备----')
                for sh in self.shs:
                    sh.executeGsctl('restart', '',
                                    param='-M standby',
                                    get_detail=True)

            self.log.info("----校验主备关系是否正常----")
            host_name = self.nodes[0].sh('hostname').result()
            self.log.info(host_name)
            if self.shs[0].check_whether_need_switch(host_name):
                f_msg = self.Constant.FAILOVER_SUCCESS_MSG
                self.log.info('----所有节点执行降备以免出现多主----')
                for sh in self.shs:
                    sh.executeGsctl('restart', '',
                                    param='-M standby',
                                    get_detail=True)
                try:
                    self.log.info('----备1升主----')
                    result = self.shs[1].executeGsctl('failover', f_msg)
                    assert result is True
                    self.log.info("----重建备机----")
                    self.shs[1].get_standby_and_build()
                except Exception as e:
                    self.log.info('备1升主失败！' + str(e))
                    self.log.info('----备2升主----')
                    result = self.shs[2].executeGsctl('failover', f_msg)
                    assert result is True
                    self.log.info("----重建备机----")
                    self.shs[2].get_standby_and_build()
                finally:
                    self.log.info('----备升主、重建完成----')
                result = self.shs[0].exec_refresh_conf()
                self.log.info(result)
                self.log.info('---进行主备切换---')
                self.shs[0].executeGsctl('switchover', '',
                                         get_detail=True)
                result = self.shs[0].exec_refresh_conf()
                self.log.info(result)

            self.log.info("----校验备机是否需要重建----")
            if self.shs[0].check_whether_need_build():
                self.shs[0].get_standby_and_build()

            db_status = self.shs[0].getDbClusterStatus('status')
            if not db_status:
                raise Exception('数据库状态异常，自动恢复失败，请检查！')

            func(*args, **kwargs)

        return wrapper
